package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class LogSplitProcessFuncation extends ProcessFunction<String,String> {
    OutputTag<String> errorLog;
    OutputTag<String> startLog;
    OutputTag<String> displayLog;
    OutputTag<String> actionLog;

    public LogSplitProcessFuncation(OutputTag<String> errorLog, OutputTag<String> startLog, OutputTag<String> displayLog, OutputTag<String> actionLog) {
        this.errorLog = errorLog;
        this.startLog = startLog;
        this.displayLog = displayLog;
        this.actionLog = actionLog;
    }

    @Override
        public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
            JSONObject jsonObject = JSON.parseObject(value);
            // 判断是否为错误日志
            if (null != jsonObject.get("err")) {
                ctx.output(errorLog, jsonObject.toJSONString());
                jsonObject.remove("err");
            }

            //  判断是否为启动日志
            if (null != jsonObject.get("start")) {
                ctx.output(startLog, jsonObject.toJSONString());
            }

            else {
                JSONArray displays = jsonObject.getJSONArray("displays");
                JSONArray action = jsonObject.getJSONArray("actions");
                jsonObject.remove("displays");
                jsonObject.remove("actions");

                out.collect(jsonObject.toJSONString());
                // 曝光日志
                if (null != displays && !displays.isEmpty()) {
                    for (int i = 0; i < displays.size(); i++) {
                        JSONObject displayJsonobject = displays.getJSONObject(i);
                        jsonObject.put("display", displayJsonobject);
                        ctx.output(displayLog, jsonObject.toJSONString());
                    }
                    jsonObject.remove("display");
                }
                // 动作日志
                if (null != action && !action.isEmpty()) {
                    for (int i = 0; i < action.size(); i++) {
                        jsonObject.put("action", action.getJSONObject(i));
                        ctx.output(actionLog, jsonObject.toJSONString());
                    }
                }
            }
        }

}

